#include <workflow/Workflow.h>
#include <workflow/WFTaskFactory.h>
#include <workflow/MySQLResult.h>
#include <protocol/frpc_msgio.h>
#include "mysql_handler.h"
#include <cppbase/logging.h>
#include <cppbase/utility.h>
#include <cppbase/cstring.h>
#include <filestore/file_store.h>
//#include <fcntl.h>// open
#include <fstream> //lseek
#include "handlers.h"
namespace Frpc
{
    Router get_default_router()
    {
        Router _router;
        _router.add_handler(uint32_t(Command::Hello), Hello);
        _router.add_handler(uint32_t(Command::HandShake), HandShake);
        _router.add_handler(uint32_t(Command::LoginUser), LoginUser);
        _router.add_handler(uint32_t(Command::LoginToken), LoginToken);
        _router.add_handler(uint32_t(Command::GetCollectionList), GetCollectionList);
        _router.add_handler(uint32_t(Command::GetCollectionContent), GetCollectionContent);
        _router.add_handler(uint32_t(Command::GetContractBar), GetContractBar);

        return _router;
    }
    std::string make_select(std::string table, std::string fields, std::string where)
    {
        std::string ans = "select " + fields + " from " + table;
        if (where.size() > 0)
            return ans + " where";
        else
            return ans;
    }

    void Hello(Request *req, Reply *rep, Context *ctx)
    {

        rep->set_ok_output("Hi");
    }
    void NoCMD(Request *req, Reply *rep, Context *ctx)
    {
        rep->set_bad_output(Status::NotFound, std::string("Handler not implemented"));
    } //<END>
    void HandShake(Request *req, Reply *rep, Context *ctx)
    {
        DLOG("call hand shake handerl.");
        rep->set_ok_output("1234567812345678");
    }
    void LoginUser(Request *req, Reply *rep, Context *ctx)
    {
        DLOG("call login user.");
        ctx->user_id = 1;
        rep->set_ok_output("welcome");

    } //<END>
    void LoginToken(Request *req, Reply *rep, Context *ctx)
    {
        //DLOG("in logintoken handler");
        // if (strcmp((char *)"token_test", req->get_content()) == 0)
        if (req->input() == "token_test")
        {
            rep->set_ok_output("Welcome");
            ((Context *)ctx)->user_id = 1;
        }
        else
        {

            rep->set_bad_output(Status::Unauthorized, "Login failed");
        }
    }

    void GetCollectionList(Request *req, Reply *rep, Context *ctx)
    {
        int user_id = ((Context *)ctx)->user_id;

        switch (uint8_t(*req->input().data()))
        {
        case 0: //request anyone collection and private collection
        {
            auto next_task = WFTaskFactory::create_mysql_task(AppGlobal::instance()->get_config("mysql"), MYSQL_RETRY, DBHandler::reply_collections);
            next_task->get_req()->set_query("select id,name,type,about from collection_info where owner_id=1 or owner_id=" + std::to_string(user_id));
            next_task->user_data = rep;
            //should cache result. so that determine auth later
            //  ((Context *)ctx)->work->push_back(next_task);
        }
        break;

        default:
            break;
        }
    }

    void GetCollectionContent(Request *req, Reply *rep, Context *ctx)
    {
        DLOG("call GetCollectionContent.");
        Frpc::Args args;
        auto s = req->input();
        auto res = args.ParseFromArray(req->input().data(), req->input().size());
        if (!res)
        {
            DLOG("failed to parse Args.");
            rep->set_bad_output(Frpc::Status::BadRequest);
            return;
        }
        if (args.arg_int64_size() != 1)
        {
            DLOG("bad request: args.arg_int64_size() != 1");
            rep->set_bad_output(Frpc::Status::BadRequest);
            return;
        }
        int collection_id = args.arg_int64(0);
        auto cache_file = AppGlobal::instance()->get_config("cache_dir") + "/mysql/" + std::to_string(int(Command::GetCollectionContent)) + "_" + std::to_string(collection_id);
        std::ifstream ifs;
        //DLOG("%s",cache_file.data());
        ifs.open(cache_file, std::ios::in);
        if (ifs.is_open())
        {
            ifs.seekg(0, std::ios::end);
            auto size = ifs.tellg();
            ifs.seekg(0, std::ios::beg);
            char *buf = (char *)malloc(size);
            ifs.read(buf, size);
            rep->set_ok_output();
            rep->add_output_nocopy(buf, size);
            Frpc::ContractArray ca;
            ca.ParseFromArray(buf, size);
            DLOG("server side contract size:%d", ca.contract_size());
        }
        else
        {
            DLOG("failed to open cache file,collection id:%d", collection_id);
            rep->set_bad_output(Status::InternalServerError);
        }
    } //<END>

    void GetContractBar(Request *req, Reply *rep, Context *ctx)
    {
        DLOG("call GetContractBar.");
        Frpc::GetBarRequest r;
        auto in = req->input();
        if (!r.ParseFromArray(in.data(), in.size()))
        {
            DLOG("failed to parse GetContractBar.");
            return;
        }
        auto barpath = AppGlobal::instance()->get_config("bar_path");
        if (barpath.size() == 0)
        {
            FATAL("unknown bar path");
        }
        MyFileStore fstore(barpath);
        int exch = int(r.exchange_id());
        auto symbol = r.symbol();
        int start_date = r.start_date();
        std::string_view res;
        auto bs = r.barsize();
        int rowsize;
        int rowcnt;
        if (start_date != 0)
        {
            res = fstore.get_daybar(::Exchange(exch), symbol, BarSize(bs), start_date, r.end_date(), &rowsize, &rowcnt);
        }
        else
        {
            int rbegin = r.rbegin();
            int count = r.count();
            res = fstore.get_rbar_series(::Exchange(exch), symbol, BarSize(bs), rbegin, count, &rowsize, &rowcnt);
        }
        rep->set_ok_output(uint16_t(rowsize), uint16_t(rowcnt)); //rowsize(1 byte),row count(2bytes)
        rep->add_output_nocopy((void *)res.data(), res.size());
    } //<END>
    void UpdateContractBar(Request *req, Reply *rep, Context *ctx)
    {
        DLOG("call UpdateContractBar.");
        auto in = req->input();
        auto barpath = AppGlobal::instance()->get_config("bar_path");
        if (barpath.size() == 0)
        {
            FATAL("unknown bar path");
        }
        // exchange_id,symbol,rowsize,rowcnt
        CString cs(rep->input());
        int e=0;
        int exch=cs.read_int16(&e);//first 2 bytes 
        std::string symbol=std::string(cs.read_strview(&e));
        BarSize bs=BarSize(cs.read_int8(&e));
        int rowsize=cs.read_int16(&e);
        int rowcnt=cs.read_int16(&e);//fllowing 2 bytes 

        if(e!=0 || req->input().size()-cs.tellg()!=rowcnt*rowsize)
        {
            rep->set_bad_output(Status::BadRequest,"failed to read content bytes");
            return;
        }

        MyFileStore fstore(barpath);
        bool res;
        if(bs==BarSize::Day1)
            res = fstore.update_daybar(::Exchange(exch), symbol, (Bar*)cs.current(),cs.unread_size()/sizeof(Bar));
        else
            DLOG("not implement");
        if(res)
            rep->set_ok_output("saved"); //rowsize(1 byte),row count(2bytes)
        else
            rep->set_bad_output(Status::InternalServerError,"failed to save file");
        
    }
}
